package prictice.transaction1;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.transactional.ITransactionalSpout;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.tuple.Values;

import java.math.BigInteger;
import java.util.Map;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 20:04
 * @Email elonyong@163.com
 */
public class MyEmitter implements ITransactionalSpout.Emitter<MyMate> {

    Map<Long, String> dbMap;

    public MyEmitter(Map<Long, String> dbMap) {
        this.dbMap = dbMap;
    }

    @Override
    public void emitBatch(TransactionAttempt transactionAttempt, MyMate myMate,
                          BatchOutputCollector batchOutputCollector) {
        long beginPoint = myMate.getBeginPoint();
        int num = myMate.getNum();
        for (long i = beginPoint; i < num + beginPoint; i++) {
            if (dbMap.get(i) == null) {
                break;
            }
            batchOutputCollector.emit(new Values(transactionAttempt, dbMap.get(i)));
        }
    }

    @Override
    public void cleanupBefore(BigInteger bigInteger) {

    }

    @Override
    public void close() {

    }
}
